package com.suxin.netty.remoting;

import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;

import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author Tang
 * @classname AbstractNettyRemoting
 * @description [  ]
 * @date 2022/12/11 16:13
 */
@Slf4j
public abstract class AbstractNettyRemoting {

    /**
     * 公用线程池
     */
    protected ExecutorService publicExecutor;

    /**
     * 默认的processor
     */
    protected Pair<NettyRequestProcessor, ExecutorService> defaultProcessor;

    /**
     * 请求处理table
     * <code,processor>
     */
    protected final HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>> processorTable = new HashMap<>(8);

    static {
        // 定时检查
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
        scheduledExecutorService.schedule(ResponseSynchronizationManager::removeResponseCallback, 1L, TimeUnit.SECONDS);
    }

    public AbstractNettyRemoting() {
    }

    /**
     * DefaultProcessor
     *
     * @param processor
     * @param executor
     */
    public void registerDefaultProcessor(NettyRequestProcessor processor, ExecutorService executor) {
        this.defaultProcessor = new Pair<>(processor, executor);
    }

    /**
     * registerProcessor
     *
     * @param code
     * @param processor
     * @param executor
     */
    public void registerProcessor(int code, NettyRequestProcessor processor, ExecutorService executor) {
        if (executor == null) {
            executor = this.publicExecutor;
        }
        processorTable.put(code, new Pair<>(processor, executor));
    }

    /**
     * 处理请求&响应
     *
     * @param ctx ctx
     * @param msg msg
     */
    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) {
        RemotingCommand cmd = msg;
        if (cmd != null) {
            switch (cmd.getType()) {
                case REQUEST_COMMAND:
                    processRequestCommand(ctx, cmd);
                    break;
                case RESPONSE_COMMAND:
                    processResponseCommand(ctx, cmd);
                    break;
                default:
                    break;
            }
        }
    }

    /**
     * process Request
     *
     * @param ctx ctx
     * @param msg msg
     */
    protected void processRequestCommand(ChannelHandlerContext ctx, RemotingCommand msg) {
        final int code = msg.getCode();
        Pair<NettyRequestProcessor, ExecutorService> servicePair = this.processorTable.get(msg.getCode());
        Pair<NettyRequestProcessor, ExecutorService> pair = servicePair == null ? defaultProcessor : servicePair;
        final String opaque = msg.getOpaque();
        if (pair != null) {
            Runnable runnable = () -> {
                // 处理器处理完成后，默认的处理结果的回调
                RemotingResponseCallback callback = (response -> {
                    response.makeResponse();
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                });
                // 异步：异步处理业务，处理完在回调
                if (pair.getObject1() instanceof AsyncNettyRequestProcessor) {
                    AsyncNettyRequestProcessor processor = (AsyncNettyRequestProcessor) pair.getObject1();
                    processor.AsyncProcessRequest(ctx, msg, callback);
                } else {
                    // 同步：处理器处理完，在返回结果
                    NettyRequestProcessor processor = pair.getObject1();
                    RemotingCommand response = processor.processRequest(ctx, msg);
                    callback.callback(response);
                }
            };
            // 提交给线程池执行
            pair.getObject2().submit(runnable);
        } else {
            // 异常回复
            String error = " request type " + code + " not supported";
            RemotingCommand response = RemotingCommand.createResponseCommand(
                    RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error
            );
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            log.error(error);
        }
    }

    /**
     * process Response
     *
     * @param ctx ctx
     * @param msg msg
     */
    private void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand msg) {
        // 唯一标识
        String opaque = msg.getOpaque();
        ResponseFuture responseFuture = ResponseSynchronizationManager.getCallback(opaque);
        if (responseFuture != null) {
            ResponseSynchronizationManager.removeCallback(opaque);
            // 执行注册的回调
            responseFuture.executeInvokeCallback();
        }
    }
}
